package com.wyt.mq.kafka;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Properties;

/**
 * @ClassName Consumer
 * @Description: TODO
 * @Author wangyongtao
 * @Date 2021/7/25 12:12
 * @Version 1.0
 **/
public class KafkaConsumerDemo {
    private final static Logger logger = LoggerFactory.getLogger(KafkaConsumerDemo.class);

    /**
     * 1、消费者：从kafka上接收对象消息，将json字符串转化为对象，便于获取消息的时候可以使用get方法获取。
     * @TODO 可以自定义继承KafkaConsumer+实现Consumer接口，这样可以控制在kafka消费过程中做些自定义处理
     */
    @Test
    public void getMsgFromKafka() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");//服务器ip:端口号，集群用逗号分隔
        props.put("group.id", "test3");  // 该字段用于区分消费者，如果存在多个消费者，group.id不能一样，如果一样，只能保证其中一个消费者消费到数据。
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        Consumer<String,String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("news_access")); //TOPIC_NAME
        while (true) {
            ConsumerRecords<String,String> records = consumer.poll(100);
            if (records.count() > 0) {
                for (ConsumerRecord<String,String> record : records) {
                    logger.info("从kafka接收到的消息是：" + record.value());
                    System.out.println("从kafka接收到的消息是：" + record.value());
                }
            }
        }
    }
}